package xyz.wpxy.iotlink.network.mqtt.server.service;

import org.eclipse.paho.client.mqttv3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class MqttBrokerConsume implements MqttCallback {

    @Autowired
    private MqttMessageProcess mqttMessageProcess;

    private final Logger logger = LoggerFactory.getLogger(MqttBrokerConsume.class);
    private MqttClient client;
    private String[] topics;


    public MqttBrokerConsume() {
    }

    public MqttBrokerConsume(MqttClient client, String[] topics) {
        this.client = client;
        this.topics = topics;
    }

    public void setClient(MqttClient client, String[] topics) {
        this.client = client;
        this.topics = topics;
    }

    //连接丢失
    @Override
    public void connectionLost(Throwable throwable) {
        logger.error("连接丢失 {}", throwable.getMessage());
        int res = 1;
        int timeout = 1;
        while (true) {
            logger.warn("mqtt-border间隔{}(s)重连{}次 {}", timeout, res, client.getClientId());

            try {
                Thread.sleep(timeout * 1000);
            } catch (Exception e) {
                logger.error("mqtt-boker延时{}(s)", timeout, e);
            }

            try {
                client.reconnect();
            } catch (MqttException e) {
                logger.error("mqtt-border 重连失败{} {}", client.getCurrentServerURI(), client.getClientId(), e);
            }
            if (client.isConnected()) {
                logger.debug("mqtt-border重连成功 {}次 {}", res, client.getClientId());
                try {
                    if (topics == null || topics.length == 0) {
                        break;
                    }
                    client.subscribe(topics);
                } catch (MqttException e) {
                    logger.error("", e);
                }
                break;
            }
            //最大重连间隔不超过60秒
            timeout = res % 60 + 1;
            res++;
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) {
        String content = new String(mqttMessage.getPayload());
        logger.info("mqtt收到消息：topic:{},Qos:{},msg:{}", topic, mqttMessage.getQos(), content);

        try {
            //TODO 消息处理
            mqttMessageProcess.receive(topic,content);
        } catch (Exception e) {
            logger.error("mqtt-broker数据处理失败{} {}", topic, mqttMessage.getId(), e);
        }
    }

    //传输完成
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        logger.debug("传输完成");
    }
}
